package com.shujia.window

import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import java.time.Duration

object Demo05CarProcess {
  /**
   * 决定堵车的两个因素：
   * 1、车流量
   * 2、车的平均速度
   * 实现的方式：
   * 1、使用Flink进行实时统计
   * 2、使用数据里自带的时间（事件时间）
   * 3、使用滑动窗口（每隔1分钟统计最近10分钟的数据）
   * 4、返回：卡口，时间，车流量，车的平均速度
   * (Long,Long,Int,Double)
   */
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    // 设置Flink任务的时间属性为 事件时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val carDS: DataStream[(Long, Long, Double)] = env.socketTextStream("master", 8888)
      .map(jsonStr => {
        val jsonObj: JSONObject = JSON.parseObject(jsonStr)
        val card: Long = jsonObj.getLong("card")
        val time: Long = jsonObj.getLong("time") * 1000
        val speed: Double = jsonObj.getDouble("speed")
        (card, time, speed)
      })

    val assignCarDS: DataStream[(Long, Long, Double)] = carDS.assignTimestampsAndWatermarks(
      WatermarkStrategy
        // 指定水位线往前移5s
        .forBoundedOutOfOrderness(Duration.ofSeconds(5))
        .withTimestampAssigner(new SerializableTimestampAssigner[(Long, Long, Double)] {
          // 从数据中提取出事件时间
          override def extractTimestamp(element: (Long, Long, Double), recordTimestamp: Long): Long = {
            element._2
          }
        }))

    assignCarDS
      .keyBy(_._1)
      // 每隔1分钟统计最近10分钟的卡口数据
      .timeWindow(Time.minutes(10), Time.minutes(1))
      .process(new ProcessWindowFunction[(Long, Long, Double), (Long, Long, Int, Double), Long, TimeWindow] {
        override def process(key: Long, context: Context, elements: Iterable[(Long, Long, Double)], out: Collector[(Long, Long, Int, Double)]): Unit = {
          // 统计车流量
          val cnt: Int = elements.size
          // 统计平均速度
          val avgSpeed: Double = elements.map(_._3).sum / cnt
          // 以窗口的结束时间最为统计时间
          val time: Long = context.window.getEnd
          out.collect((key, time, cnt, avgSpeed))
        }
      }).print()


    env.execute()


  }

}
